package com.turing.model.registry.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.turing.common.util.ObjectUtils;
import com.turing.common.util.StringUtils;
import com.turing.common.util.Tools;
import com.turing.model.registry.dao.RegisterInstanceMapper;
import com.turing.model.registry.inner.XxlJobAdminInnerService;
import com.turing.model.registry.model.domain.RegisterInstance;
import com.turing.model.registry.service.RegisterInstanceService;
import com.turing.model.registry.vo.RegisteredInstanceVO;
import com.turing.model.registry.vo.XxlJobInfoVo;
import com.xxl.job.core.biz.model.ReturnT;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;

/**
* @Description 服务类实现
* @Author livin
* @Date 2022-01-24
*/
@Service
public class RegisterInstanceServiceImpl extends ServiceImpl<RegisterInstanceMapper, RegisterInstance> implements RegisterInstanceService {

    private static  Log logger = LogFactory.getLog(RegisterInstanceServiceImpl.class);

    private Integer duration = 180;

    @Value("${analysis.threads.count}")
    private int analysisThreadsCount;

    @Autowired
    private XxlJobAdminInnerService xxlJobAdminInnerService;

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void registerOrRenew(RegisteredInstanceVO registeredInstanceVO) {

        String host = registeredInstanceVO.getHost();
        String port = registeredInstanceVO.getPort();
        String appType = registeredInstanceVO.getApp_type();
        Map<String, Integer> incomingGraphicsInfos = registeredInstanceVO.getGpu_info();
        Map<String, String[]> gpuElementMap = registeredInstanceVO.getGpu_element_map();
        List<RegisterInstance> currentGraphicsInfos = lambdaQuery().eq( RegisterInstance::getHost, host )
                .eq( RegisterInstance::getPort, port )
                .eq( RegisterInstance::getAppType, appType ).list();
        // 拼接显卡 场景信息用作重复显卡场景筛选
        List<String> currentCardsAndTemplates = currentGraphicsInfos.stream().map(s -> s.getCardNo() + "_" + s.getTemplates()).collect(Collectors.toList());
        List<String> registerCardsAndTemplates = new ArrayList<>();
        for (Map.Entry<String, String[]> entry : gpuElementMap.entrySet()) {
            registerCardsAndTemplates.add(entry.getKey() + "_" + StringUtils.join(entry.getValue(), ","));
        }
        // 不存在该机器的注册信息，新建定时任务以及写入注册信息
        if(Tools.isEmpty(currentGraphicsInfos)){
            for (Map.Entry<String, Integer> entry : incomingGraphicsInfos.entrySet()) {
                String incomingCardNo = entry.getKey();
                // 检查显卡是否可用，模型实例是否不存在
                if(entry.getValue() == 0 && !instanceExist(host, port, appType, incomingCardNo)){
                    String[] templates = gpuElementMap.get(incomingCardNo);
                    String templateStr = StringUtils.join(templates,",");
                    Integer jobId = addJob(host, port, appType, incomingCardNo, templateStr);
                    saveInstanceInfo(host, port, appType, incomingCardNo, jobId, templateStr);
                    startJob(jobId);
                }
            }

        }else{
            // 如果gpu信息相等，只更新时间
            if(registerCardsAndTemplates.equals(currentCardsAndTemplates)){
                currentGraphicsInfos.forEach(cgi->cgi.setExpirationTime(LocalDateTime.now().plusSeconds(duration)));
                currentGraphicsInfos.forEach(this::updateById);
            }else{
                // 处理重复的注册信息
                List<String> incomingCopy = new ArrayList<>();
                if(Tools.isNotEmpty(registerCardsAndTemplates)){
                    incomingCopy = (List<String>)ObjectUtils.deepClone(registerCardsAndTemplates);
                    assert incomingCopy != null;
                    incomingCopy.retainAll(currentCardsAndTemplates);
                    registerCardsAndTemplates.removeAll(incomingCopy);
                    currentCardsAndTemplates.removeAll(incomingCopy);
                }
                // 重复注册，做更新处理
                if(Tools.isNotEmpty(incomingCopy)){
                    incomingCopy.forEach(ic->{
                        String cardNo = ic.split("_")[0];
                        updateById(RegisterInstance.builder().host(host).port(port).appType(appType).cardNo(cardNo).build());
                    });
                }
                // 移除未注册的模型以及任务
                if(Tools.isNotEmpty(currentCardsAndTemplates)){
                    currentCardsAndTemplates.forEach(
                        cgi->{
                            String cardNo = cgi.split("_")[0];
                            // remove job
                            stopAndRemoveJob(host, port, appType, cardNo);
                            // remove instance not exist
                            removeInstanceInfo(host, port, appType, cardNo);
                        }
                    );
                }
                // 注册新的模型以及任务
                if(Tools.isNotEmpty(registerCardsAndTemplates)){
                    registerCardsAndTemplates.forEach(
                        igi->{
                            String cardNo = igi.split("_")[0];
                            if(!instanceExist(host, port, appType, cardNo)) {
                                String[] templates = gpuElementMap.get(cardNo);
                                String templateStr = StringUtils.join(templates,",");
                                Integer jobId = addJob(host, port, appType, cardNo, templateStr);
                                saveInstanceInfo(host, port, appType, cardNo, jobId, templateStr);
                                startJob(jobId);
                            }
                        }
                    );
                }
            }
        }
        if(appType.equals("0")){
            registerOrRenewAnalysis(host,port);
        }
    }

    @Override
    public List<RegisterInstance> getExpiredInstances() {
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        String format = LocalDateTime.now().format(formatter);
        LambdaQueryWrapper<RegisterInstance> queryWrapper = new LambdaQueryWrapper<RegisterInstance>()
                .apply("UNIX_TIMESTAMP(expiration_time) < UNIX_TIMESTAMP('" + format + "')");
                //.ge(RegisterInstance::getExpirationTime,new Date());
        return list(queryWrapper);
    }

    @Override
    public void saveInstanceInfo(String host, String port, String appType, String cardNo,Integer taskId, String templates) {

        RegisterInstance ri = RegisterInstance.builder().host(host).port(port).appType(appType).cardNo(cardNo).jobId(taskId).templates(templates)
            .expirationTime(LocalDateTime.now().plusSeconds(duration)).build();
        save(ri);
    }

    @Override
    public void removeInstanceInfo(String host, String port, String appType, String cardNo) {
        lambdaUpdate().eq(RegisterInstance::getHost,host)
                .eq(RegisterInstance::getPort,port)
                .eq(RegisterInstance::getAppType,appType)
                .eq(RegisterInstance::getCardNo,cardNo)
                .set(RegisterInstance::getIsDel,1).update();
    }

    @Override
    public Integer addJob(String host, String port, String appType, String cardNo, String templates){
        XxlJobInfoVo xxlJobInfoVo = XxlJobInfoVo.builder()
                .jobGroup(0)
                .author("livin")
//                .alarmEmail("873502652@qq.com")
                .scheduleType("CRON")
                .scheduleConf("*/1 * * * * ?")
                .glueType("BEAN")
                .misfireStrategy("DO_NOTHING")
                .executorBlockStrategy("SERIAL_EXECUTION")
                .executorTimeout(5000)
                .executorFailRetryCount(0)
                .executorRouteStrategy("LEAST_FREQUENTLY_USED")
                .glueRemark("GLUE代码初始化")
                .glueSource(null)
                .build();
        // 目前只有预测模型需要根据场景进行筛选，ocr（all）、解析（null）不做处理
        if ("all".equals(templates) || templates == "") {
            xxlJobInfoVo.setExecutorParam(host+":"+port+"_"+cardNo);
        } else {
            xxlJobInfoVo.setExecutorParam(host+":"+port+"_"+cardNo + "_" + templates);
        }
        String jobHandler = getJobHandler(appType);
        String jobDesc = getJobDesc(appType);
        xxlJobInfoVo.setExecutorHandler(jobHandler);
        xxlJobInfoVo.setJobDesc(jobDesc);
        ReturnT<Integer> res = xxlJobAdminInnerService.getIdByAppName("model-caller");
        if(res.getCode() == ReturnT.SUCCESS_CODE){
            xxlJobInfoVo.setJobGroup(res.getContent());
            ReturnT<String> returnT = xxlJobAdminInnerService.addJob(xxlJobInfoVo);
            if(returnT.getCode() == ReturnT.SUCCESS_CODE ){
                return Integer.parseInt(returnT.getContent());
            }else{
                return 0;
            }
        }else{
            return 0;
        }

    }

    private String getJobHandler(String appType) {

        switch (appType){
            case "0":
                return "callOcrJobHandler";
            case "1":
                return "callForecastJobHandler";
            case "2":
                return "callAnalysisJobHandler";
            default:
                return "";
        }

    }

    private String getJobDesc(String appType) {

        switch (appType){
            case "0":
                return "ocr任务";
            case "1":
                return "预测任务";
            case "2":
                return "解析任务";
            default:
                return "";
        }

    }

    @Override
    public void stopAndRemoveJob(String host, String port, String appType, String cardNo){

        List<RegisterInstance> registerInstances = lambdaQuery().eq(RegisterInstance::getHost,host)
                .eq(RegisterInstance::getPort,port)
                .eq(RegisterInstance::getAppType,appType)
                .eq(RegisterInstance::getCardNo,cardNo).last("limit 1").list();
        if(Tools.isNotEmpty(registerInstances)){
            Integer jobId = registerInstances.get(0).getJobId();
            xxlJobAdminInnerService.stopJob(jobId);
            xxlJobAdminInnerService.removeJob(XxlJobInfoVo.builder().id(jobId).build());
        }

    }

    @Override
    public void stopJob(Integer jobId) {

        xxlJobAdminInnerService.stopJob(jobId);

    }

    @Override
    public void startJob(Integer jobId) {

        xxlJobAdminInnerService.startJob(jobId);

    }

    @Override
    public boolean instanceExist(String host, String port, String type, String cardNo) {
        return lambdaQuery().eq(RegisterInstance::getHost,host)
                .eq(RegisterInstance::getPort,port)
                .eq(RegisterInstance::getAppType,type)
                .eq(RegisterInstance::getCardNo,cardNo).count() > 0;
    }

    private void registerOrRenewAnalysis(String host, String port) {

        List<RegisterInstance> currentAnalysisInfos = lambdaQuery().eq( RegisterInstance::getHost, host )
                .eq( RegisterInstance::getPort, port )
                .eq( RegisterInstance::getAppType, "2" ).list();
        if(Tools.isEmpty(currentAnalysisInfos)){
            for (int i = 0; i < analysisThreadsCount; i++) {
                Integer jobId = addJob(host, port, "2", "0", "");
                saveInstanceInfo(host, port, "2", "0",jobId,"");
                startJob(jobId);
            }
        }else{
            currentAnalysisInfos.forEach(cai->cai.setExpirationTime(LocalDateTime.now().plusSeconds(duration)));
            currentAnalysisInfos.forEach(this::updateById);
        }


    }


}
